package com.foreveross.fetl.queue;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.inject.Inject;

import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.apache.tools.zip.ZipOutputStream;
import org.springframework.stereotype.Component;

import com.common.remote.ftp.FtpConfig;
import com.common.remote.sftp.SFtpFileTransferClient;
import com.dayatang.domain.InstanceFactory;
import com.foreveross.fetl.application.ExcelInfoApplication;
import com.foreveross.fetl.application.ITaskTimeApplication;
import com.foreveross.fetl.common.cfg.system.ReportProperties;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.common.vo.hangxin.PlaneInfo;
import com.foreveross.fetl.common.vo.model.doub.DoublePlaneInfo;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.HangxinDateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.AirLinesData;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.JsonUtil;
import com.foreveross.fetl.util.PlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;
import com.foreveross.fetl.util.ZipUtils;
import com.foreveross.fetl.vo.ExcelInfoVO;
import com.foreveross.fetl.vo.TaskTimeVO;

@Component
public class MetadataForAirsDoublePreprocess {
	private static Logger log = Logger.getLogger("metadataForAirs");
	
	@Inject
	private ExcelInfoApplication excelInfoApplication;
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	private String taskTime=null;
	
	private StringBuffer keys=new StringBuffer();
	private StringBuffer results=new StringBuffer();
	
	/**
	 * 初始化计数，添加日志
	 */
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		keys.setLength(0);
		results.setLength(0);
		LogMessageUtils.create("航信元数据", this.getClass().getSimpleName());
	}

	
	public void execute(){
		Calendar calendar=Calendar.getInstance();
		int hour =calendar.get(Calendar.HOUR_OF_DAY);
		List<TaskTimeVO> listTask=InstanceFactory.getInstance(ITaskTimeApplication.class).find(null, FetlErrorUtils.metadataForAirs, hour);
		if(null!=listTask && listTask.size()>0){
			rollBackNumber();
			Date sTime=null;
			Date eTime=null;
			String dateStr=DateUtils.format(new Date(), "yyyyMMdd");
			String start=null;
			String end=null;
			for(TaskTimeVO vo:listTask){
				if(vo.getGrabStart()<10)
					start=dateStr+"0"+vo.getGrabStart();
				else
					start=dateStr+vo.getGrabStart();
				if(vo.getGrabStop()<10)
					end=dateStr+"0"+vo.getGrabStop();
				else
					end=dateStr+vo.getGrabStop();
				sTime=DateUtils.parse(start, "yyyyMMddHH");
				eTime=DateUtils.parse(end, "yyyyMMddHH");
				List<String> lines=AirLinesData.getDoubleMapAirLines().get("hangxin").get(vo.getChannel());
				Map<String,List<String>> hangxinMap=new HashMap<String,List<String>>();
				hangxinMap.put(vo.getChannel(), lines);
				Map<String,Map<String,List<String>>> map =new HashMap<String,Map<String,List<String>>>();
				map.put("hangxin", hangxinMap );
				process(sTime,eTime,map);
			}
			LogMessageUtils.build("航信元数据", this.getClass().getSimpleName(), 2, results.toString(), 
					failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime,keys.toString());
			MetadataCountPreprocess count=new MetadataCountPreprocess();
			count.countforExcel(calendar.getTime());
		}
	}
	
	

	public void manualExecute(final String start,String time){
		this.rollBackNumber();
		Map<String,Map<String,List<String>>> map =AirLinesData.getDoubleMapAirLines();
		try {
		List<String> timeList = Arrays.asList(time.split(";"));
		taskTime = start + " " +time;
		String str =start;
		str += timeList.get(0);
		Date startTime = DateUtils.parse(str, "yyyyMMddHH");
		Date endTime=DateUtils.parse(start+timeList.get(timeList.size()-1), "yyyyMMddHH");
		this.process(startTime, endTime,map);
		LogMessageUtils.build("航信元数据", this.getClass().getName(), 2, results.toString(), 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime,keys.toString());
		} catch (Exception e) {
			log.error(StringUtils.getStackTrace(e));
			LogMessageUtils.build("航信元数据", this.getClass().getName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime,keys.toString());
		}
	}
	
	
	public void hangxinExcute(String start,String end ,String channel){
		this.rollBackNumber();
		try {
			Date startTime = DateUtils.parse(start, "yyyyMMddHH");
			Date endTime = DateUtils.parse(end, "yyyyMMddHH");
			if("all".equals(channel)){
				Map<String,Map<String,List<String>>> map =AirLinesData.getDoubleMapAirLines();
				process(startTime,endTime,map);
			}else{
				List<String> lines=AirLinesData.getDoubleMapAirLines().get("hangxin").get(channel);
				Map<String,List<String>> hangxinMap=new HashMap<String,List<String>>();
				hangxinMap.put(channel, lines);
				Map<String,Map<String,List<String>>> map =new HashMap<String,Map<String,List<String>>>();
				map.put("hangxin", hangxinMap );
				process(startTime,endTime,map);
			}
			LogMessageUtils.build("航信元数据", this.getClass().getName(), 2, results.toString(), 
					failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime,keys.toString());
		} catch (Exception e) {
			log.error(StringUtils.getStackTrace(e));
			LogMessageUtils.build("航信元数据", this.getClass().getName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime,keys.toString());
		}
		
	}

	private void process(Date time,Date endTime,Map<String,Map<String,List<String>>> map) {
		log.error(DateUtils.format(time, "yyyyMMddHH"));
		log.error(DateUtils.format(endTime, "yyyyMMddHH"));
		//重新初始化汇率
		ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));
		File files =null;
		Map<String,String> fileNames=new HashMap<String,String>();
		for(String companyName:map.keySet()){
			Map<String,List<String>> airLines=map.get(companyName);
			for(String hbaseKey:airLines.keySet()){
				tasksNumber++;
			   	this.findFormHbase(hbaseKey,airLines.get(hbaseKey),time,endTime); // 创建文件
				 files=new File(ReportProperties.getFilePath()+"company/newFile/"+hbaseKey);
				 File[] list =files.listFiles();
				 if(null!=list && list.length>0){
					 	FileOutputStream fous=null;
						ZipOutputStream zipOut =null;
						File uploadFile=null;
						File file =null;
						try {
							uploadFile= new File(ReportProperties.getFilePath()+"company/"+hbaseKey+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+".zip");
							if (!uploadFile.exists()) {
								uploadFile.createNewFile(); 
							}
							fous=new FileOutputStream(uploadFile);
							zipOut =new ZipOutputStream(fous);
							 for (int i=0; i<list.length; i++) {
					               if (list[i].isFile()) {
					            	   if(list[i].getName().indexOf(DateUtils.format(endTime, "yyyy-MM-dd"))<=0){
					            		  continue;
					            	   }
					            	   file=list[i];
					            	   ZipUtils.zipFile(file,zipOut); 
					            	   file=null; 
					               }
					              
						       }
							zipOut.flush();
							zipOut.close();
							zipOut=null;
							fous.flush();
							fous.close();
							boolean success=false;
							try {
								success=this.uploadToHangxin(uploadFile);
								results.append(uploadFile.getName()+" "+ uploadFile.length() + "<br>");
							} catch (Exception e) {
								success=this.uploadToHangxin(uploadFile);
							}
							if(!success){
								results.append(uploadFile.getName()+" upload error"+  "<br>");
							}
							fileNames.put(hbaseKey, uploadFile.getPath());
						//	uploadFile.delete();
							ZipUtils.delAllFile(files.getPath());
						}catch (Exception e) {
							failuresNumber++;
							results.append(uploadFile.getName()+" upload error"+ StringUtils.getStackTrace(e) + "<br>");
							log.error(StringUtils.getStackTrace(e));
							FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
									this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
							FetlErrorUtils.saveToHbase(vo);
							
						}finally {
							if(null!=zipOut){
								try {
									zipOut.flush();
									zipOut.close();
								} catch (IOException e) {
									e.printStackTrace();
								}
							}
							if(null!=fous){
								try {
									fous.flush();
									fous.close();
								} catch (IOException e) {
									e.printStackTrace();
								}
							}
						}	
				 }
			}
			for(String key:fileNames.keySet()){
				File file =null;
				try {
					 file =new File(fileNames.get(key));
					this.uploadFile(file, key, endTime, companyName);
					file.delete();
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}finally{
					if(null!=file && file.exists()){
						file.delete();
					}
				}
			}
		}
		
		
	}
	
	/**
	 * 从hbase里读数据，并生成txt文件
	 * @param source
	 * @param lines
	 * @param time
	 */
	private void findFormHbase(String source, List<String> lines, Date time,Date endTime) {
		ResultScanner rs = null;
		int thousand=0;
		String path=ReportProperties.getFilePath()+"company/newFile/"+source+"/";
		Calendar cStart = Calendar.getInstance();
		cStart.setTime(time);
		Calendar cEnd = Calendar.getInstance();
		cEnd.setTime(endTime);
		int i=1;
		int count=0;
		String id=null;
		keys.append("渠道： "+ source);
		StringBuilder josnString=new StringBuilder(1024*1024);
		String  separator=System.getProperty("line.separator");
		List<String> listkeys=new ArrayList<String>();
		Set<Date> flightDateSet=HangxinDateUtil.getFlightDateSet();
		Integer[] returnDays=HangxinDateUtil.getReturnDays();
		while (cStart.before(cEnd)) {
			String timeStr=DateUtils.format(cStart.getTime(), "yyyyMMddHH");
			for(String line:lines){
				String[] citys=line.split("-");
				for(Date flightDate:flightDateSet){
					String goDateStr=DateUtils.format(flightDate, "yyyyMMdd");
					for(int j=0;j<returnDays.length;j++){
						Calendar c =Calendar.getInstance();
						c.setTime(flightDate);
						c.add(Calendar.DAY_OF_MONTH, returnDays[j]);
						Date returnFlightDate=c.getTime();
						String returnDay=DateUtils.format(returnFlightDate, "yyyyMMdd");
						String startRow=RowkeyUtils.getRowKey(timeStr, source,citys[0],citys[1],goDateStr,returnDay, RowkeyUtils.START_ROW_KEY);
						String endRow=RowkeyUtils.getRowKey(timeStr, source, citys[0],citys[1],goDateStr,returnDay, RowkeyUtils.STOP_ROW_KEY);
						try {
							try {
								rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
										startRow, endRow,new String[]{"planeinfo","transitsInfo","cabinsInfo","stopOversInfo",
										"returnPlaneInfosInfo","returnCabinsInfo","returnStopOversInfo","returnTransitsInfo"},null);
								rs.iterator();
							} catch (Exception e) {
								rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
										startRow, endRow,new String[]{"planeinfo","transitsInfo","cabinsInfo","stopOversInfo",
										"returnPlaneInfosInfo","returnCabinsInfo","returnStopOversInfo","returnTransitsInfo"},null);
								//如果从hbase里读数据失败了就再来一次
							}
								for (Result res : rs) {
									try {
										Map<String, Map<String, String>> curMap = HBaseHelper
												.transformHBaseRowToMap(res);
										Map<String, String> curPalneInfoMap = curMap.get("planeinfo");
										
										id =new String(res.getRow());
										listkeys.add(id);
										if(i==1){
											keys.append(" 开始rowkey： "+ id+"<br>");
											i++;
										}
										DoublePlaneInfo planeInfo= null;
										double refePrice=ExchangeRateUtil.getRefePrice(curPalneInfoMap.get("currency"));
										if(curPalneInfoMap.get("grabChannelNum").equals("00006") && null==curPalneInfoMap.get("currency")){
											refePrice=ExchangeRateUtil.getRefePrice("USD");
										}
										planeInfo=HbaseToPlaneInfoUtil.hbaseToPlaneInfo(curMap, refePrice,DoublePlaneInfo.class);
										if(null!=planeInfo && null!=planeInfo.getFlightNo()){
											List<PlaneInfo> list2=PlaneInfoUtil.changeToPlaneInfo(planeInfo);
											if(null!=list2 && list2.size()>0){
												for(PlaneInfo vo:list2){
													josnString.append(JsonUtil.getJSONString(vo));
													josnString.append(separator);
													count++;
												}
											}
										}
										if(count>=5000){
											thousand++;
											count=0;
											String fileName=source+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
											ZipUtils.creatFile(path, fileName, josnString.toString());
											josnString.delete(0, josnString.length());
											listkeys.clear();
										}
									} catch (Exception e) {
										FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
												this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
										vo.setKeys(listkeys);
										FetlErrorUtils.saveToHbase(vo);
										listkeys.clear();
										count=0;
									}
								}
						} catch (Exception e) {
							log.error(StringUtils.getStackTrace(e));
							FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
									this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
							vo.setStartRowKey(startRow);
							vo.setEndRowkey(endRow);
							FetlErrorUtils.saveToHbase(vo);
						}finally{
							if(rs != null) rs.close();
							rs = null;
						}
					}
				}
			}
			cStart.add(Calendar.HOUR_OF_DAY, 1);// 开始时间加一小时
		}
		try {
			if(count>0){
				thousand++;
				String fileName=source+"_"+DateUtils.format(endTime, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
				ZipUtils.creatFile(path, fileName, josnString.toString());
				josnString.delete(0, josnString.length());
				listkeys.clear();
			}
		} catch (Exception e) {
			FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
					this.getClass().getName(),DateUtils.format(endTime, "yyyyMMddHH"), e);
			vo.setKeys(listkeys);
			FetlErrorUtils.saveToHbase(vo);
			listkeys.clear();
		}
		if(i>1){
			keys.append(" 结束rowkey： "+ id +"<br>");	
		}else{
			keys.append(" 没数据 <br> ");
		}
		
	}

	/**
	 * 将文件上传到Hangxin FTP 上
	 * @param f 文件
	 * @return
	 */
	private boolean uploadToHangxin(File f){
		String address =ReportProperties.getHangxinAddress();
		int port= Integer.parseInt(null==ReportProperties.getHangxinPort()? "21":ReportProperties.getHangxinPort());
		String userName= ReportProperties.getHangxinUserName();
		String passWord =ReportProperties.getHangxinPassword();
		String path=ReportProperties.getHangxinPath();
		if(!path.endsWith("/")) path+="/";
		boolean success = false; 
		try {
			Long times=System.currentTimeMillis();
			if("/".equals(path)){
				success=this.uploadFile(address, port, userName, passWord, null, f);	
			}else{
				success=this.uploadFile(address, port, userName, passWord, path, f);
			}
			log.error(f.getName());
			log.error(DateUtils.timeInMillis2Time(System.currentTimeMillis()-times));
		}catch (Exception e) {
			try {
				throw e;
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		}finally{
			
		}
		return success;
	}
	
	/**
	 * 将文件上传到 我们的SFTP上
	 * @param f
	 * @param hbasekey
	 * @param time
	 * @param type
	 */
	private void uploadFile(File f, String hbasekey, Date time,String type) {
		String address =ReportProperties.getAddress();
		int port= Integer.parseInt(null==ReportProperties.getPort()? "22":ReportProperties.getPort());
		String userName= ReportProperties.getUserName();
		String passWord =ReportProperties.getPassword();
		String path=ReportProperties.getPath();
		if(!path.endsWith("/")) path+="/";
		String dateType =DateUtils.format(time, "yyyyMMddHH");
		path=path.replace("webdata/", "");
		String dir=path +type+"/round-trip/"+dateType.substring(0, 4)+"/"+dateType.substring(4, 6)+"/"+dateType.substring(6, 8)+"/"+hbasekey;
		//ftp连接配置
    	FtpConfig config =new FtpConfig(address,port,userName,passWord);
		SFtpFileTransferClient sftpClient=new SFtpFileTransferClient(config);
		InputStream in =null;
		boolean flag=false;
		try {
			try {
				flag = sftpClient.connect();
			} catch (Exception e) {
				log.error(StringUtils.getStackTrace(e));
				Thread.sleep(10 * 1000L);
				flag = sftpClient.connect();
			}

			if(flag){
				//如果目录不存在则创建
				String[] dirs=dir.split("/");
				StringBuffer buffer=new StringBuffer();
				for(String d:dirs){
					buffer.append(d+"/");
					if(!sftpClient.isDirectoryExists(buffer.toString())){
						sftpClient.makeDirectory(buffer.toString());
					}
				}
				in=new FileInputStream(f);
				
				sftpClient.upload(in, dir+"/"+f.getName());
				
				sftpClient.disconnect();
				insertNumber++;
				ExcelInfoVO oldVo =excelInfoApplication.findByName(f.getName(),type,2);
				if(null!=oldVo){
					excelInfoApplication.delete(oldVo);
				}
				ExcelInfoVO excelInfoVO =new ExcelInfoVO();
				excelInfoVO.setCode(hbasekey);
				excelInfoVO.setDateType(dateType);
				excelInfoVO.setType(2);
				excelInfoVO.setFileType(type);
				excelInfoVO.setName(f.getName());
				excelInfoVO.setPath(f.getName());
			//	excelInfoVO.setPath(type+"/round-trip/"+dateType.substring(0, 4)+"/"+dateType.substring(4, 6)+"/"+dateType.substring(6, 8)+"/"+hbasekey+"/"+f.getName());
				excelInfoVO.setUploadDate(new Date());
				excelInfoApplication.saveOrUpdate(excelInfoVO);
		//		in.close();
			}
		} catch (Exception e) {
			try {
				throw e;
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		}finally{
			if(null!=sftpClient){
				sftpClient.disconnect();	
			}
			try {
			if (in != null) {
				in.close();
			}
		} catch (Exception e) {
		}
		}
	}

	
	/**
	 * Description: 向FTP服务器上传文件
	 * @param url FTP服务器hostname
	 * @param port FTP服务器端口
	 * @param username FTP登录账号
	 * @param password FTP登录密码
	 * @param path FTP服务器保存目录
	 * @param filename 上传到FTP服务器上的文件名
	 * @param input 输入流
	 * @return 成功返回true，否则返回false
	 */
	private  boolean uploadFile(String url,int port,String username, String password, String path, File file) {
		boolean success = false;
		FTPClient ftp = new FTPClient();
		InputStream input=null;
		try {
			int reply;
			boolean flag=false;
			input=new FileInputStream(file);
			try {
				ftp.connect(url, port);//连接FTP服务器
				//如果采用默认端口，可以使用ftp.connect(url)的方式直接连接FTP服务器
				ftp.login(username, password);//登录
				if(!flag){
					ftp.connect(url, port);
					ftp.login(username, password);
				}
			} catch (Exception e) {
				ftp.connect(url, port);//连接FTP服务器
				//如果采用默认端口，可以使用ftp.connect(url)的方式直接连接FTP服务器
				ftp.login(username, password);//登录
				if(!flag){
					ftp.connect(url, port);
					ftp.login(username, password);
				}
			}
			ftp.enterLocalPassiveMode();//设置客户端PASV模式 (客户端主动连服务器端；端口用20) 有防火墙用户不能使用主动模式 ，这是因为防火墙不允许来自网外的主动连接，所以用户必须同使用被动模 式
            ftp.setFileType(FTPClient.BINARY_FILE_TYPE);  
            ftp.setBufferSize(1024*1024*10);
			reply = ftp.getReplyCode();
			if (!FTPReply.isPositiveCompletion(reply)) {
				ftp.disconnect();
				return success;
			}
			ftp.changeWorkingDirectory(path);
			ftp.storeFile(file.getName(), input);	
			input.close();
			File flagFile=new File(file.getName().replace(".zip", "_flag"));
			if(!flagFile.exists()){
				flagFile.createNewFile();
			}
			input=new FileInputStream(flagFile);
			ftp.storeFile(flagFile.getName(), input);
			input.close();
			flagFile.delete();
			success = true;
		} catch (IOException e) {
			//log.error(StringUtils.getStackTrace(e));
			try {
				throw e;
			} catch (IOException e1) {
				e1.printStackTrace();
			}
		} finally {
			try {
				if(null!=input)
				input.close();
			} catch (IOException e) {
				log.error(StringUtils.getStackTrace(e));
			}
			try {
				ftp.logout();
			} catch (IOException e) {
				log.error(StringUtils.getStackTrace(e));
			}
			if (ftp.isConnected()) {
				try {
					ftp.disconnect();
				} catch (IOException ioe) {
					log.error(StringUtils.getStackTrace(ioe));
				}
			}
		}
		return success;
	}
}
